package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo02Map {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setAppName("map")
      .setMaster("local")

    //spark上下文数据

    val sc = new SparkContext(conf)

    /**
      * map：一行一行处理rdd中的数据
      *
      *
      */


    /**
      *  构建rdd的方法
      * 1、读取文件
      * 2、基于scala集合构建rdd
      */

    //基于scala集合构建rdd
    val listRdd: RDD[Int] = sc.parallelize(List(1,2,3,4,5,6,7,8,9))

    println("ListRdd的分区数:"+listRdd.getNumPartitions)

    val mapRDD: RDD[Int] = listRdd.map(i =>{
      i*2
    })

    /**
      * mapvalues：处理kv格式rdd的value
      */
    //转换成kv格式
    val kvRDD: RDD[(Int, Int)] = listRdd.map(i =>(i,i))

    val mapvalues: RDD[(Int, Int)] = kvRDD.mapValues(i => i*2)
    mapvalues.foreach(println)

    /**
      * mapPartitions：一次处理一个分区的数据，
      * 返回值需要时一个迭代器
      */
    listRdd.mapPartitions((iter: Iterator[Int]) => {
      val list: List[Int] = iter.toList
      list.map(i => i * 2).toIterator
    }).foreach(println)
    val mapPartitionsRDD: RDD[Int] = listRdd.mapPartitionsWithIndex{
      case(index:Int,iter: Iterable[Int]) =>
        println("当前分区编号："+index)
        iter
    }
    mapPartitionsRDD.foreach(println)
  }
}
